package com.king.app.dwm

import java.text.SimpleDateFormat

import com.alibaba.fastjson.{JSON, JSONException, JSONObject, TypeReference}
import com.fasterxml.jackson.databind.ObjectMapper
import com.king.app.function.TableProcessFunction
import com.king.bean.{OrderDetail, OrderInfo, OrderWide}
import com.king.util.MyKafkaUtil
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector
import org.slf4j.LoggerFactory

/**
 * @Author: KingWang
 * @Date: 2022/2/12  
 * @Desc:
 **/
object OrderWideApp {

  def main(args: Array[String]): Unit = {

    val log = LoggerFactory.getLogger("OrderWideApp")

    //1. 环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    //2. 读取kafka主题的数据，并转换为JSON对象，提取时间戳生成watermark
    val orderInfoSourceTopic = "dwd_order_info"
    val orderDetailSourceTopic = "dwd_order_detail"
    val orderWideSinkTopic = "dwm_order_wide"
    val groupId = "order_wide_group"
    val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
    val mapper = new ObjectMapper()

    val orderInfoStream = env.addSource(MyKafkaUtil.getKafkaConsumer(orderInfoSourceTopic, groupId))
      .map { x =>

        val orderInfo = try {

//          val obj = mapper.readValue(x,classOf[OrderInfo])
          val obj = JSON.parseObject(x, classOf[OrderInfo])  //这里注意：如果class中有BigDecimal类型的，需要使用java.math.BigDecimal

          val createTime = obj.create_time
          obj.setCreate_date(createTime.substring(0, 10))
          obj.setCreate_hour(createTime.substring(12, 13))
          obj.setCreate_ts(sdf.parse(createTime).getTime)
          obj
        } catch {
          case e:Exception => {
            log.error("JSON转换OrderInfo异常:" + x)
            null
          }
        }
        orderInfo
      }.filter(x=> null != x).assignTimestampsAndWatermarks(WatermarkStrategy.forMonotonousTimestamps[OrderInfo]()
      .withTimestampAssigner(new SerializableTimestampAssigner[OrderInfo] {
        override def extractTimestamp(element: OrderInfo, recordTimestamp: Long): Long = element.create_ts
      }))

    val orderDetailStream = env.addSource(MyKafkaUtil.getKafkaConsumer(orderDetailSourceTopic, groupId))
      .map { x =>

        val orderDetail = try {

//          val obj = mapper.readValue(x,classOf[OrderDetail])
          val obj = JSON.parseObject(x, classOf[OrderDetail])  //这里注意：如果class中有BigDecimal类型的，需要使用java.math.BigDecimal

          val createTime = obj.create_time
          obj.setCreate_ts(sdf.parse(createTime).getTime)
          obj
        } catch {
          case e: Exception => {
            e.printStackTrace()
            log.error("JSON转换OrderDetail异常:" + x)
            null
          }
        }
        orderDetail
      }.filter(x => null != x)
      .assignTimestampsAndWatermarks(WatermarkStrategy.forMonotonousTimestamps[OrderDetail]()
        .withTimestampAssigner(new SerializableTimestampAssigner[OrderDetail] {
          override def extractTimestamp(element: OrderDetail, recordTimestamp: Long): Long = element.create_ts
        }))


    //3. 双流join
    val joinStream = orderInfoStream.keyBy(_.id).intervalJoin(orderDetailStream.keyBy(_.order_id))
      .between(Time.seconds(-10), Time.seconds(10))
      .process(new ProcessJoinFunction[OrderInfo, OrderDetail, OrderWide] {
        override def processElement(orderInfo: OrderInfo, orderDetail: OrderDetail, ctx: ProcessJoinFunction[OrderInfo, OrderDetail, OrderWide]#Context, out: Collector[OrderWide]): Unit = {
          out.collect(new OrderWide(orderInfo, orderDetail))
        }
      })

    joinStream.map(x=> JSON.toJSON(x).toString).print("orderWideWithNoDimDS>>>>>>>>")

    //4. 关联维度信息
    //5. 将数据写入kafka

    //6. 启动
    env.execute("order_wide_app")
  }

}
